package yunjiao.javatutorials.guava.concurrent.comprehensive;

import lombok.Getter;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 数据库连接池监控
 *
 * @author yangyunjiao
 */
public class MonitoredConnectionPool {
    private final Lock lock = new ReentrantLock();
    private final ComprehensiveCondition connectionAvailable;
    private final ComprehensiveCondition poolNotFull;
    private final Queue<MockConnection> availableConnections;
    private final Set<MockConnection> usedConnections;
    private final int maxConnections;
    private volatile boolean shutdown = false;

    // 连接获取统计
    private final AtomicInteger successfulAcquisitions = new AtomicInteger(0);
    private final AtomicInteger failedAcquisitions = new AtomicInteger(0);
    private final AtomicInteger totalWaitTimeMs = new AtomicInteger(0);

    public MonitoredConnectionPool(int maxConnections) {
        this.maxConnections = maxConnections;
        this.availableConnections = new LinkedList<>();
        this.usedConnections = new HashSet<>();
        this.connectionAvailable = new ComprehensiveCondition(lock.newCondition());
        this.poolNotFull = new ComprehensiveCondition(lock.newCondition());

        // 初始化连接
        for (int i = 0; i < maxConnections; i++) {
            availableConnections.offer(new MockConnection("Connection-" + (i + 1)));
        }
    }

    public MockConnection getConnection(long timeout, TimeUnit unit)
            throws InterruptedException, TimeoutException {
        if (shutdown) {
            throw new IllegalStateException("连接池已关闭");
        }

        long startTime = System.currentTimeMillis();

        lock.lock();
        try {
            long nanosTimeout = unit.toNanos(timeout);

            while (availableConnections.isEmpty() && !shutdown) {
                if (nanosTimeout <= 0) {
                    failedAcquisitions.incrementAndGet();
                    throw new TimeoutException("获取连接超时");
                }
                nanosTimeout = connectionAvailable.awaitNanos(nanosTimeout);
            }

            if (shutdown) {
                throw new IllegalStateException("连接池已关闭");
            }

            MockConnection connection = availableConnections.poll();
            usedConnections.add(connection);

            successfulAcquisitions.incrementAndGet();
            long waitTime = System.currentTimeMillis() - startTime;
            totalWaitTimeMs.addAndGet((int) waitTime);

            System.out.printf("获取连接成功: %s, 等待时间: %d ms%n",
                    connection.getName(), waitTime);
            System.out.printf("连接池状态: %d/%d (使用/总数)%n",
                    usedConnections.size(), maxConnections);

            return connection;

        } finally {
            lock.unlock();
        }
    }

    public void releaseConnection(MockConnection connection) {
        lock.lock();
        try {
            if (usedConnections.remove(connection)) {
                availableConnections.offer(connection);

                // 通知等待连接的线程
                connectionAvailable.signal();

                System.out.printf("释放连接: %s%n", connection.getName());
                System.out.printf("连接池状态: %d/%d (使用/总数)%n",
                        usedConnections.size(), maxConnections);
            }
        } finally {
            lock.unlock();
        }
    }

    public void shutdown() {
        lock.lock();
        try {
            shutdown = true;
            connectionAvailable.signalAll(); // 唤醒所有等待的线程

            // 关闭所有连接
            for (MockConnection conn : availableConnections) {
                conn.realClose();
            }
            for (MockConnection conn : usedConnections) {
                conn.realClose();
            }

            availableConnections.clear();
            usedConnections.clear();

        } finally {
            lock.unlock();
        }

        printConnectionPoolStatistics();
    }

    public void printConnectionPoolStatistics() {
        System.out.println("\n" + "=".repeat(70));
        System.out.println("                   数据库连接池统计报告");
        System.out.println("=".repeat(70));

        System.out.printf("连接池状态: %s%n", shutdown ? "已关闭" : "运行中");
        System.out.printf("连接统计: %d/%d (使用/总数)%n", usedConnections.size(), maxConnections);
        System.out.printf("获取统计: %d 成功, %d 失败%n",
                successfulAcquisitions.get(), failedAcquisitions.get());

        int totalAcquisitions = successfulAcquisitions.get() + failedAcquisitions.get();
        double successRate = totalAcquisitions == 0 ? 0 :
                (double) successfulAcquisitions.get() / totalAcquisitions * 100;
        System.out.printf("成功率: %.1f%%%n", successRate);

        if (successfulAcquisitions.get() > 0) {
            double avgWaitTime = (double) totalWaitTimeMs.get() / successfulAcquisitions.get();
            System.out.printf("平均等待时间: %.2f ms%n", avgWaitTime);
        }

        System.out.println("\n连接可用条件统计:");
        connectionAvailable.printDetailedStatistics();

        System.out.println("\n性能分析:");
        double connectionUtilization = (double) usedConnections.size() / maxConnections * 100;
        System.out.printf("连接利用率: %.1f%%%n", connectionUtilization);

        double waitToSuccessRatio = successfulAcquisitions.get() == 0 ? 0 :
                (double) connectionAvailable.getWaitCount() / successfulAcquisitions.get();
        System.out.printf("等待/成功比例: %.2f%n", waitToSuccessRatio);

        System.out.println("=".repeat(70));
    }

    // 模拟数据库连接
    public static class MockConnection {
        @Getter
        private final String name;
        private volatile boolean closed = false;

        public MockConnection(String name) {
            this.name = name;
        }

        public void executeQuery(String sql) {
            if (closed) {
                throw new IllegalStateException("连接已关闭");
            }
            System.out.printf("%s 执行查询: %s%n", name, sql);
        }

        public void close() {
            // 实际应用中这里会返回连接池
            System.out.printf("%s 返回到连接池%n", name);
        }

        public void realClose() {
            closed = true;
            System.out.printf("%s 真正关闭%n", name);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MonitoredConnectionPool pool = new MonitoredConnectionPool(3);

        // 创建多个线程模拟并发获取连接
        List<Thread> threads = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            final int threadId = i;
            Thread thread = new Thread(() -> {
                try {
                    MockConnection conn = pool.getConnection(2, TimeUnit.SECONDS);

                    // 模拟使用连接
                    Thread.sleep(500 + (int)(Math.random() * 1000));
                    conn.executeQuery("SELECT * FROM table_" + threadId);

                    // 释放连接
                    pool.releaseConnection(conn);

                } catch (Exception e) {
                    System.out.println("线程 " + threadId + " 错误: " + e.getMessage());
                }
            });
            threads.add(thread);
        }

        // 启动所有线程
        for (Thread thread : threads) {
            thread.start();
            Thread.sleep(100); // 错开启动时间
        }

        // 等待所有线程完成
        for (Thread thread : threads) {
            thread.join();
        }

        // 关闭连接池并打印统计
        pool.shutdown();
    }
}
